package feign.reactive;

import feign.InvocationHandlerFactory.MethodHandler;
import feign.Target;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscription;

public abstract class ReactiveInvocationHandler implements InvocationHandler {

    private final Target<?> target;
    private final Map<Method, MethodHandler> dispatch;

    public ReactiveInvocationHandler(Target<?> target, Map<Method, MethodHandler> dispatch) {
        this.target = target;
        this.dispatch = dispatch;
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) {
        if ("equals".equals(method.getName())) {
            try {
                Object otherHandler = args.length > 0 && args[0] != null ? Proxy.getInvocationHandler(args[0]) : null;
                return equals(otherHandler);
            } catch (IllegalArgumentException e) {
                return false;
            }
        } else if ("hashCode".equals(method.getName())) {
            return hashCode();
        } else if ("toString".equals(method.getName())) {
            return toString();
        }
        return this.invoke(method, this.dispatch.get(method), args);
    }

    @Override
    public int hashCode() {
        return this.target.hashCode();
    }

    @Override
    public boolean equals(Object obj) {
        if (obj == null) {
            return false;
        }
        if (obj == this) {
            return true;
        }
        if (ReactiveInvocationHandler.class.isAssignableFrom(obj.getClass())) {
            return this.target.equals(obj);
        }
        return false;
    }

    @Override
    public String toString() {
        return "Target [" + this.target.toString() + "]";
    }

    protected abstract Publisher invoke(Method method, MethodHandler methodHandler, Object[] arguments);

    Publisher<?> invokeMethod(MethodHandler methodHandler, Object[] arguments) {
        return subscriber -> subscriber.onSubscribe(new Subscription() {
            private final AtomicBoolean isTerminated = new AtomicBoolean(false);

            @Override
            public void request(long n) {
                if (n <= 0 && !terminated()) {
                    subscriber.onError(new IllegalArgumentException("negative subscription request"));
                }
                if (!isTerminated()) {
                    try {
                        Object result = methodHandler.invoke(arguments);
                        if (null != result) {
                            subscriber.onNext(result);
                        }
                    } catch (Throwable th) {
                        if (!terminated()) {
                            subscriber.onError(th);
                        }
                    }
                }
                if (!terminated()) {
                    subscriber.onComplete();
                }
            }

            @Override
            public void cancel() {
                isTerminated.set(true);
            }

            private boolean isTerminated() {
                return isTerminated.get();
            }

            private boolean terminated() {
                return isTerminated.getAndSet(true);
            }
        });
    }
}
